package com.qyer.search.engine;

import static com.google.common.collect.Maps.newHashMapWithExpectedSize;
import static com.qyer.search.Constants.INDEX_NAME;
import static com.qyer.search.Constants.INDEX_TYPES;
import static com.qyer.search.model.lastminute.SKU.RETAIN_FIELDS;
import static com.qyer.search.model.lastminute.SKU.RETAIN_FIELDS_BASIC;

import com.google.common.collect.Sets;
import com.qyer.log.DebugLogger;
import com.qyer.search.Constants;
import com.qyer.search.exception.SearchException;
import com.qyer.search.model.SearchParameter;
import com.qyer.search.model.SearchParameterEntity;
import com.qyer.search.model.SearchParameterOrigin;
import com.qyer.search.model.lastminute.ZSKU;
import com.qyer.search.searchplan.SQPlan;
import com.qyer.search.searchplan.SQPlan2;
import com.qyer.search.searchplan.SQPlan3;
import org.apache.commons.configuration.CompositeConfiguration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.admin.cluster.stats.ClusterStatsRequest;
import org.elasticsearch.action.admin.cluster.stats.ClusterStatsResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.AdminClient;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.ClusterAdminClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.threadpool.ThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.qyer.search.enums.*;

/**
 * User: Z J Wu Date: 2016/01/21 Time: 10:16 Package: com.qyer.search.engine
 */
public abstract class AbstractESEngine {

  private static final Logger LOGGER = LoggerFactory.getLogger(AbstractESEngine.class);

  protected Client client;
  protected Node node;
  protected String clientType;
  protected final String CLIENT_TYPE_TRANSPORT = "transport";
  protected final String CLIENT_TYPE_NODE = "node";
  protected String indexName;
  //+++++++++++++++++

  protected void buildClient() {
    CompositeConfiguration c = new CompositeConfiguration();
    try {
      c.addConfiguration(new PropertiesConfiguration("search_config.properties"));
    } catch (ConfigurationException e) {
      e.printStackTrace();
    }
    String host = c.getString("search.engine.hosts", "es.qyer.com");
    String[] hosts = host.split("\\|");
    Set<String> set = Sets.newHashSet(hosts);
    indexName = c.getString("search.engine.index","products");
    System.out.println("Index:" + indexName);
    int port = c.getInt("search.engine.port", 9300);
    String clusterName = c.getString("search.engine.clusterName");
    this.clientType = c.getString("search.engine.clientType", "transport");
    Settings settings = Settings.settingsBuilder().put("cluster.name", clusterName).build();
    try {
      if (CLIENT_TYPE_TRANSPORT.equals(this.clientType)) {
        TransportClient tc = TransportClient.builder().settings(settings).build();
        System.out.println("\n\n\n~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
        Iterator<ThreadPool.Info> infoIterator = tc.threadPool().info().iterator();
        while (infoIterator.hasNext()){
          ThreadPool.Info info = infoIterator.next();
          System.out.println("Name:" + info.getName());
          System.out.println("Min:" + info.getMin());
          System.out.println("Max:" + info.getMax());
          if(info.getQueueSize() != null)
            System.out.println("Queue:" + info.getQueueSize().toString());
          if(info.getKeepAlive() != null)
            System.out.println("KeepAlive:" + info.getKeepAlive().toString());
          if(info.getThreadPoolType() != null)
            System.out.println("PoolType:" + info.getThreadPoolType().getType());
        }
        System.out.println("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n\n\n");
        for (String s : set) {
          LOGGER.info("ElasticSearchVar - host added - {}", s);
          tc.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(s), port));
        }
        client = tc;
      } else if (CLIENT_TYPE_NODE.equals(this.clientType)) {
        node = NodeBuilder.nodeBuilder().settings(settings).build();
        client = node.client();
      } else {
        throw new Exception("Client type is invalid - " + this.clientType);
      }
      LOGGER.info("ElasticSearchVar - port - {}", port);
      LOGGER.info("ElasticSearchVar - cluster-name - {}", clusterName);
      LOGGER.info("ElasticSearchVar - status - {}", status());
    } catch (Exception e) {
      LOGGER.error("Cannot build transport client.", e);
    }
  }

  public ClusterHealthStatus status() {
    AdminClient adminClient = client.admin();
    ClusterAdminClient cac = adminClient.cluster();
    ClusterStatsResponse response = cac.clusterStats(new ClusterStatsRequest()).actionGet();
    return response.getStatus();
  }

  public void shutdown() {
    if (CLIENT_TYPE_TRANSPORT.equals(this.clientType) && this.client != null) {
      this.client.close();
    } else if (CLIENT_TYPE_NODE.equals(this.clientType) && this.node != null) {
      this.node.close();
    }
    LOGGER.info("ES client is shutdown.");
  }

  public static void setHighlight(SearchRequestBuilder srb) {
    srb.addHighlightedField("title");
    srb.setHighlighterPreTags("<u>");
    srb.setHighlighterPostTags("</u>");
  }

  public static void logDebugMap(SearchParameter sp, boolean cacheHit, SQPlan plan) {
    Map<String, Object> map = newHashMapWithExpectedSize(2);
    map.put("hit", cacheHit);
    map.put("plan_id", plan.getId());
    map.put("query", plan.getQueryBuilder());
    DebugLogger.logDebug(sp, map);
  }

  public static void logDebugMap(SearchParameter sp, boolean cacheHit, SQPlan2 plan) {
    Map<String, Object> map = newHashMapWithExpectedSize(2);
    map.put("hit", cacheHit);
    map.put("plan_id", plan.getId());
    map.put("query", plan.getQueryBuilder());
    DebugLogger.logDebug(sp, map);
  }


  public static void logDebugMap(SearchParameter sp, boolean cacheHit, QueryBuilder qb) {
    Map<String, Object> map = newHashMapWithExpectedSize(2);
    map.put("hit", cacheHit);
    map.put("query", qb);
    DebugLogger.logDebug(sp, map);
  }



  public SearchRequestBuilder prepareSearch() {
    //return client.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPES);
    return client.prepareSearch(this.indexName).setTypes(INDEX_TYPES);
  }

  public List<String> executeExtractAsString(SearchParameter sp, QueryBuilder qb,
                                             boolean showHighlight, boolean simpleResult) {
    SearchResponse sr = execute(sp, qb, showHighlight, simpleResult);
    SearchHits hits = sr.getHits();
    if (hits == null) {
      return null;
    }
    SearchHit[] hitArray = hits.getHits();
    if (ArrayUtils.isEmpty(hitArray)) {
      return null;
    }
    String str;
    List<String> results = new ArrayList<>(hitArray.length);
    for (SearchHit hit : hitArray) {
      str = hit.getSourceAsString();
      if (StringUtils.isNotBlank(str)) {
        results.add(str);
      }
    }
    return results;
  }

  public List<String> executePlanExtractAsString(SearchParameter sp, SQPlan plan,
                                                 boolean showHighlight, boolean simpleResult) {
    SearchResponse sr = executePlan(sp, plan, showHighlight, simpleResult);
    SearchHits hits = sr.getHits();
    if (hits == null) {
      return null;
    }
    SearchHit[] hitArray = hits.getHits();
    if (ArrayUtils.isEmpty(hitArray)) {
      return null;
    }
    String str;
    List<String> results = new ArrayList<>(hitArray.length);
    for (SearchHit hit : hitArray) {
      str = hit.getSourceAsString();
      if (StringUtils.isNotBlank(str)) {
        results.add(str);
      }
    }
    return results;
  }

  //线上使用
  public List<String> executePlanExtractAsString(SearchParameter sp, SQPlan plan,
                                                 boolean showHighlight, boolean simpleResult, boolean basic) {
    //测试
    //SearchResponse sr = executePlan(sp, plan, showHighlight, simpleResult, basic);
    SearchResponse sr = executePlanEx(sp, plan, showHighlight, simpleResult, basic);
    SearchHits hits = sr.getHits();
    //System.out.println("Total hits:" + hits.getTotalHits());
    //System.out.println("Max  Score:" + hits.getMaxScore());
    if (hits == null) {
      return null;
    }
    SearchHit[] hitArray = hits.getHits();
    if (ArrayUtils.isEmpty(hitArray)) {
      return null;
    }
    String str;
    List<String> results = new ArrayList<>(hitArray.length);
    for (SearchHit hit : hitArray) {
      str = hit.getSourceAsString();
      if (StringUtils.isNotBlank(str)) {
        results.add(str);
      }
    }
    return results;
  }

  public Set<String> executePlanExtractAsSet(SearchParameter sp, SQPlan2 plan,
                                                 boolean showHighlight, FieldsMode mode) {
    //测试
    //System.out.println("GET HITS go");
    //SearchResponse sr = executePlan(sp, plan, showHighlight, simpleResult, basic);
    long t1 = System.currentTimeMillis();
    SearchResponse sr = executePlanEx(sp, plan, showHighlight, mode);
    SearchHits hits = sr.getHits();
    //System.out.println("hitsgetTime:"+(System.currentTimeMillis() - t1));
    //System.out.println("线程ID:" + Thread.currentThread() + "###搜索时间:" + sr.getTookInMillis());
    if (hits == null) {
      return null;
    }
    long t5 = System.currentTimeMillis();
    SearchHit[] hitArray = hits.getHits();
    //System.out.println("gethitsTime:"+(System.currentTimeMillis() - t5));
    if (ArrayUtils.isEmpty(hitArray)) {
      return null;
    }
    String str;
    int siezl = 0;
    int size1 = 0;
    int size2 = 0;
    Set<String> results = new HashSet<>((int)hits.getTotalHits());
    long t3 = System.currentTimeMillis();
    for (SearchHit hit : hitArray) {
      size1++;
      str = hit.getSourceAsString();
      if (StringUtils.isNotBlank(str)) {
        results.add(str);
        size2++;
      }else {
        siezl++;
      }
    }
    long t4 = System.currentTimeMillis();
    /*System.out.println("for time:"+(t4 - t3));
    System.out.println("ATOMIC all SIZE:"+size1);
    System.out.println("ATOMIC is SIZE:"+size2);
    System.out.println("ATOMIC RESULT SIZE:"+results.size());
    System.out.println("ATOMIC null SIZE:" + siezl);
    System.out.println("executePlanExtractAsSet time:"+(System.currentTimeMillis() - t1));*/

    long t11 = System.currentTimeMillis();
    /*for (SearchHit hit : hitArray) {
      hit.getSource();
    }*/
    //System.out.println("getJacksonTime:"+(System.currentTimeMillis()- t11));

    return results;
  }

  public Set<String> executePlanExtractAsIDSet(SearchParameter sp, SQPlan3 plan,
                                             boolean showHighlight, FieldsMode mode) {
    //测试
    //System.out.println("GET HITS go");
    //SearchResponse sr = executePlan(sp, plan, showHighlight, simpleResult, basic);
    long t1 = System.currentTimeMillis();
    SearchResponse sr = executePlanEx(sp, plan, showHighlight, mode);
    SearchHits hits = sr.getHits();
    //System.out.println("hitsgetTime:"+(System.currentTimeMillis() - t1));
    //System.out.println("线程ID:" + Thread.currentThread() + "###搜索时间:" + sr.getTookInMillis());
    if (hits == null) {
      return null;
    }
    long t5 = System.currentTimeMillis();
    SearchHit[] hitArray = hits.getHits();
    //System.out.println("gethitsTime:"+(System.currentTimeMillis() - t5));
    if (ArrayUtils.isEmpty(hitArray)) {
      return null;
    }
    String str;
    int siezl = 0;
    int size1 = 0;
    int size2 = 0;
    Set<String> results = new HashSet<>((int)hits.getTotalHits());
    long t3 = System.currentTimeMillis();
    for (SearchHit hit : hitArray) {
      size1++;
      //str = hit.getSource().get(ZSKU.FIELD_ID).toString();
      str = hit.getId();
      if (StringUtils.isNotBlank(str)) {
        results.add(str);
        //size2++;
      }else {
        //siezl++;
      }
      if("EMPTY".equals(str)){
        System.out.println("ES EMPTY");
      }
    }
    long t4 = System.currentTimeMillis();
    /*System.out.println("for time:"+(t4 - t3));
    System.out.println("ATOMIC all SIZE:"+size1);
    System.out.println("ATOMIC is SIZE:"+size2);
    System.out.println("ATOMIC RESULT SIZE:"+results.size());
    System.out.println("ATOMIC null SIZE:" + siezl);
    System.out.println("executePlanExtractAsSet time:"+(System.currentTimeMillis() - t1));*/

    long t11 = System.currentTimeMillis();
    /*for (SearchHit hit : hitArray) {
      hit.getSource();
    }*/
    //System.out.println("getJacksonTime:"+(System.currentTimeMillis()- t11));

    return results;
  }

  @Deprecated public SearchResponse execute(SearchParameter sp, QueryBuilder qb,
                                            boolean showHighlight, boolean simpleResult) {
    logDebugMap(sp, false, qb);
    SearchRequestBuilder srb = prepareSearch();
    srb.setQuery(qb);
    if (simpleResult) {
      srb.setFetchSource(RETAIN_FIELDS, null);
    }
    srb.setTimeout(TimeValue.timeValueMillis(3000));
    //srb.setSearchType(DFS_QUERY_THEN_FETCH);
    srb.setSize(sp.getPageInfo().getPageSize());
    if (showHighlight) {
      setHighlight(srb);
    }
    return srb.execute().actionGet();
  }

  public SearchResponse executePlan(SearchParameter sp, SQPlan plan, boolean showHighlight,
                                    boolean simpleResult) {
    logDebugMap(sp, false, plan);
    SearchRequestBuilder srb = prepareSearch();
    srb.setQuery(plan.getQueryBuilder());
    if (simpleResult) {
      srb.setFetchSource(RETAIN_FIELDS, null);
      //srb.addFields(RETAIN_FIELDS);
    }
    srb.setTimeout(TimeValue.timeValueMillis(3000));
    //srb.setSearchType(DFS_QUERY_THEN_FETCH);
    srb.setSize(sp.getPageInfo().getPageSize());
    if (showHighlight) {
      setHighlight(srb);
    }
    return srb.execute().actionGet();
  }

  //临时增加
  public SearchResponse executePlan(SearchParameter sp, SQPlan plan, boolean showHighlight,
                                    boolean simpleResult, boolean basic) {
    logDebugMap(sp, false, plan);
    SearchRequestBuilder srb = prepareSearch();
    srb.setQuery(plan.getQueryBuilder());
    if(basic){
      srb.setFetchSource(RETAIN_FIELDS_BASIC, null);
    }else if (simpleResult) {
      srb.setFetchSource(RETAIN_FIELDS, null);
    }
    srb.setTimeout(TimeValue.timeValueMillis(3000));
    srb.setSize(sp.getPageInfo().getPageSize());
    if (showHighlight) {
      setHighlight(srb);
    }
    /*
    srb.setFrom(0);
    srb.setSize(200);*/
    if(sp.isCategory()) {
      //System.out.println("人为限制！--100");
      srb.setFrom(0);
      srb.setSize(Constants.SEARCH_THRESHOLD);
    }
    return srb.execute().actionGet();
  }

  public SearchResponse executePlanEx(SearchParameter sp, SQPlan plan, boolean showHighlight,
                                    boolean simpleResult, boolean basic) {
    logDebugMap(sp, false, plan);
    SearchRequestBuilder srb = prepareSearch();
    //srb.setQuery(plan.getQueryBuilder());
    srb.setSearchType(SearchType.DFS_QUERY_AND_FETCH);
    srb.setPostFilter(plan.getQueryBuilder());
    //srb.setRequestCache(true);
    //System.out.println("Cache dis");
    if(basic){
      srb.setFetchSource(RETAIN_FIELDS_BASIC, null);
    }else if (simpleResult) {
      srb.setFetchSource(RETAIN_FIELDS, null);
    }
    srb.setTimeout(TimeValue.timeValueMillis(3000));
    srb.setSize(sp.getPageInfo().getPageSize());
    if (showHighlight) {
      setHighlight(srb);
    }
    /*if(sp.isCategory()) {
      //System.out.println("人为限制！--100");
      srb.setFrom(0);
      srb.setSize(Constants.SEARCH_THRESHOLD);
    }*/
    return srb.execute().actionGet();
  }


  public SearchResponse executePlanEx(SearchParameter sp, SQPlan2 plan, boolean showHighlight, FieldsMode mode) {

    //System.out.println("executePlanEx go~");
    long t1 = System.currentTimeMillis();
    logDebugMap(sp, false, plan);
    SearchRequestBuilder srb = prepareSearch();
    //srb.setQuery(plan.getQueryBuilder());
    srb.setPostFilter(plan.getQueryBuilder());
    srb.setFetchSource(Constants.FIELDS_MODE_MAP.get(mode), null);
    /*if(basic){
      srb.setFetchSource(RETAIN_FIELDS_BASIC, null);
    }else if (simpleResult) {
      srb.setFetchSource(RETAIN_FIELDS, null);
    }*/
    srb.setTimeout(TimeValue.timeValueMillis(3000));
    srb.setSize(5000);
    if (showHighlight) {
      setHighlight(srb);
    }
    long t2 =System.currentTimeMillis();
    //System.out.println("executePlanEx time:"+(t2-t1));
    return srb.execute().actionGet();
  }

  public SearchResponse executePlanEx(SearchParameter sp, SQPlan3 plan, boolean showHighlight, FieldsMode mode) {

    //System.out.println("executePlanEx go~");
    long t1 = System.currentTimeMillis();
    //logDebugMap(sp, false, plan);
    SearchRequestBuilder srb = prepareSearch();
    //srb.setQuery(plan.getQueryBuilder());
    srb.setFetchSource(false);
    //srb.setNoFields();
    srb.setSearchType(SearchType.QUERY_AND_FETCH);
    srb.setPostFilter(plan.getQueryBuilder());
    //srb.setFetchSource(Constants.FIELDS_MODE_MAP.get(mode), null);
    /*if(basic){
      srb.setFetchSource(RETAIN_FIELDS_BASIC, null);
    }else if (simpleResult) {
      srb.setFetchSource(RETAIN_FIELDS, null);
    }*/
    srb.setTimeout(TimeValue.timeValueMillis(3000));
    srb.setSize(5000);
    if (showHighlight) {
      setHighlight(srb);
    }
    long t2 =System.currentTimeMillis();
    //System.out.println("executePlanEx time:"+(t2-t1));
    return srb.execute().actionGet();
  }

  // For fork join search
  public List<String> forkSearch(SearchParameter searchParameter, List<SQPlan> plans,
                                  boolean showHighlight, boolean simpleResult) throws
    SearchException {
    return null;
  }

  public List<String> forkSearch(SearchParameterOrigin searchParameter, List<SQPlan> plans,
                                 boolean showHighlight, boolean simpleResult) throws
    SearchException {
    return null;
  }

  //临时增加
  public List<String> forkSearch(SearchParameterOrigin searchParameter, List<SQPlan> plans,
                                 boolean showHighlight, boolean simpleResult, boolean basic) throws
    SearchException {
    return null;
  }

  public List<String> forkSearchThread(SearchParameterOrigin searchParameter, List<SQPlan> plans,
                                 boolean showHighlight, boolean simpleResult, boolean basic) throws
    SearchException {
    return null;
  }

  public Set<String> forkSearch(SearchParameterEntity searchParameter, List<SQPlan2> plans,
                                boolean showHighlight, FieldsMode mode) throws SearchException{
    return null;
  }

  public Set<String> forkSearchThread(SearchParameterEntity searchParameter, List<SQPlan2> plans,
                                boolean showHighlight, FieldsMode mode) throws SearchException{
    return null;
  }

  public Set<String> forkSearchSimple(SearchParameterEntity searchParameter, List<SQPlan3> plans,
                                      boolean showHighlight, FieldsMode mode) throws SearchException{
    return null;
  }

}
